package com.pap.rabbitmq.trans.coordinator;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.pap.rabbitmq.data.RabbitMetaMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * 使用 redis 当做第三方的守护进程，消息的发送者和接收者之间通过redis 这个第三方守护进程进行维护
 *
 * @Auther: alexgaoyh
 * @Date: 2018/12/3 16:18
 * @Description:
 */
@Component(value = "transRabbitMQRedisCoordinator")
public class TransRabbitMQRedisCoordinator implements ICoordinator {

    /**
     * 处于ready状态消息
     */
    public static final Object MQ_MSG_READY = "trans.msg.ready";

    /**
     * 处于prepare状态消息
     */
    public static final Object MQ_MSG_PREPARE = "trans.msg.prepare";

    /**
     * 分隔符
     */
    public static final String DB_SPLIT = ",";

    /**
     * 缓存超时时间,超时进行重发
     */
    public static final long TIME_GAP = 2000;

    public static final String TIME_PATTERN = "yyyy-MM-dd HH:mm:ss";


    @Autowired
    @Qualifier(value = "transRabbitMQRedisTemplate")
    RedisTemplate redisTemplate;

    @Override
    public void setMsgPrepare(String msgId) {
        redisTemplate.opsForSet().add(MQ_MSG_PREPARE, msgId);
    }


    @Override
    public void setMsgReady(String msgId, RabbitMetaMessage rabbitMetaMessage) {
        ObjectMapper mapper = new ObjectMapper();
        try {
            String rabbitMetaMessageJSON = mapper.writeValueAsString(rabbitMetaMessage);
            redisTemplate.opsForHash().put(MQ_MSG_READY, msgId, rabbitMetaMessageJSON);
            redisTemplate.opsForSet().remove(MQ_MSG_PREPARE, msgId);
        } catch (IOException e) {
            // TODO logger 记录，人工介入
        }
    }

    @Override
    public void setMsgSuccess(String msgId) {
        redisTemplate.opsForHash().delete(MQ_MSG_READY, msgId);
    }

    @Override
    public RabbitMetaMessage getMetaMsg(String msgId) {
        return (RabbitMetaMessage) redisTemplate.opsForHash().get(MQ_MSG_READY, msgId);
    }

    @Override
    public List getMsgPrepare() throws Exception {
        SetOperations setOperations = redisTemplate.opsForSet();
        Set<String> messageIds = setOperations.members(MQ_MSG_PREPARE);
        List<String> messageAlert = new ArrayList();
        for (String messageId : messageIds) {
            /**如果消息超时，加入超时队列*/
            if (messageTimeOut(messageId)) {
                messageAlert.add(messageId);
            }
        }
        /**在redis中删除已超时的消息*/
        setOperations.remove(MQ_MSG_READY, messageAlert);
        return messageAlert;
    }

    @Override
    public List getMsgReady() throws Exception {
        HashOperations hashOperations = redisTemplate.opsForHash();
        List<RabbitMetaMessage> messages = hashOperations.values(MQ_MSG_READY);
        List<RabbitMetaMessage> messageAlert = new ArrayList();
        List<String> messageIds = new ArrayList<>();
        for (RabbitMetaMessage message : messages) {
            /**如果消息超时，加入超时队列*/
            if (messageTimeOut(message.getMessageId())) {
                messageAlert.add(message);
                messageIds.add(message.getMessageId());
            }
        }
        /**在redis中删除已超时的消息*/
        hashOperations.delete(MQ_MSG_READY, messageIds);
        return messageAlert;
    }

    @Override
    public Long incrResendKey(String key, String hashKey) {
        return redisTemplate.opsForHash().increment(key, hashKey, 1);
    }

    @Override
    public Long getResendValue(String key, String hashKey) {
        return (Long) redisTemplate.opsForHash().get(key, hashKey);
    }

    boolean messageTimeOut(String messageId) throws Exception {
        String messageTime = (messageId.split(DB_SPLIT))[1];
        long timeGap = System.currentTimeMillis() -
                new SimpleDateFormat(TIME_PATTERN).parse(messageTime).getTime();
        if (timeGap > TIME_GAP) {
            return true;
        }
        return false;
    }
}
